package com.ymm.spark.spark.page;

import com.alibaba.fastjson.JSONObject;
import com.ymm.spark.constant.Constants;
import com.ymm.spark.dao.ITaskDAO;
import com.ymm.spark.dao.factory.DAOFactory;
import com.ymm.spark.domain.PageSplitConvertRate;
import com.ymm.spark.domain.Task;
import com.ymm.spark.util.DateUtils;
import com.ymm.spark.util.NumberUtils;
import com.ymm.spark.util.ParamUtils;
import com.ymm.spark.util.SparkUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import scala.Tuple2;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 页面单跳转化率Spark作业
 *
 * @author zhangj
 * @date 2019/12/4
 */
public class PageOneStepConvertRateSpark {
	public static void main(String[] args) {
		/**
		 * 1:构建Spark上下文
		 */
		final SparkConf conf = new SparkConf().setAppName(Constants.SPARK_APP_NAME_PAGE);
		SparkUtils.setMaster(conf);

		final JavaSparkContext sc = new JavaSparkContext(conf);
		final SQLContext sqlContext = SparkUtils.getSQLContext(sc.sc());

		/**
		 * 2:生成模拟数据
		 */
		SparkUtils.mockData(sc, sqlContext);

		/**
		 * 3:查询任务获取参数
		 */
		final Long taskid = ParamUtils.getTaskIdFromArgs(args, Constants.SPARK_LOCAL_TASKID_PAGE);
		final ITaskDAO taskDAO = DAOFactory.getTaskDAO();
		final Task task = taskDAO.findById(taskid);
		if (task == null) {
			System.out.println(new Date() + ": cannot find this task with id [" + taskid + "].");
			return;
		}
		final JSONObject taskParam = JSONObject.parseObject(task.getTaskParam());

		/**
		 * 4:查询指定日期范围内的用户访问行为数据
		 */
		JavaRDD<Row> actionRDD = SparkUtils.getActionRDDByDateRange(sqlContext, taskParam);

		/**
		 * 5:返回<sessionid,row>的格式
		 */
		JavaPairRDD<String, Row> sessionid2actionRDD = getSessionid2actionRDD(actionRDD);

		/**
		 * 6:对<sessionid,row>的格式数据,groupByKey()
		 */
		final JavaPairRDD<String, Iterable<Row>> sessionid2actionsRDD = sessionid2actionRDD.groupByKey();

		/**
		 * 6:最核心的一步,每个session的单跳页面切片的生成,以及页面流的匹配算法
		 */
		JavaPairRDD<String, Integer> pageSplitRDD = generateAndMatchPageSplit(sc, sessionid2actionsRDD, taskParam);
		final Map<String, Object> pageSplitPvMap = pageSplitRDD.countByKey();

		/**
		 *7:计算页面流起始页的访问PV
		 */
		long startPagePv = getStartPagePv(taskParam, sessionid2actionsRDD);

		/**
		 * 8:计算目标页面流的各个页面切片的转化率
		 */
		Map<String, Double> convertRateMap = computePageSplitConvertRate(taskParam, pageSplitPvMap, startPagePv);

		/**
		 * 9:持久化页面切片转化率
		 */
		persistConvertRate(taskid, convertRateMap);

	}

	private static void persistConvertRate(Long taskid, Map<String, Double> convertRateMap) {
		StringBuffer stringBuffer = new StringBuffer("");
		for (Map.Entry<String, Double> convertRateEntry : convertRateMap.entrySet()) {
			String pageSplit = convertRateEntry.getKey();
			Double convertRate = convertRateEntry.getValue();
			stringBuffer.append(pageSplit + "=" + convertRate + "|");
		}

		String convertRate = stringBuffer.toString().substring(0, stringBuffer.length() - 1);

		PageSplitConvertRate pageSplitConvertRate = new PageSplitConvertRate();
		pageSplitConvertRate.setTaskid(taskid);
		pageSplitConvertRate.setConvertRate(convertRate);
		DAOFactory.getIPageSplitConvertRateDAO().insert(pageSplitConvertRate);
	}

	private static Map<String, Double> computePageSplitConvertRate(JSONObject taskParam,
	                                                               Map<String, Object> pageSplitPvMap,
	                                                               long startPagePv) {
		Map<String, Double> convertRateMap = new HashMap<String, Double>();

		String[] targetPages = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW).split(",");
		long lastPageSplitPv = 0L;
		for (int i = 1; i < targetPages.length; i++) {
			final String targetPageSplit = targetPages[i - 1] + "_" + targetPages[i];
			long targetPageSplitPv = Long.valueOf(String.valueOf(pageSplitPvMap.get(targetPageSplit)));
			double convertRate = 0.0;
			if (i == 1) {
				convertRate = NumberUtils.formatDouble((double) targetPageSplitPv / (double) startPagePv, 2);
			} else {
				convertRate = NumberUtils.formatDouble((double) targetPageSplitPv / (double) lastPageSplitPv, 2);
			}
			convertRateMap.put(targetPageSplit, convertRate);
			lastPageSplitPv = targetPageSplitPv;
		}
		return convertRateMap;
	}

	private static long getStartPagePv(JSONObject taskParam, JavaPairRDD<String, Iterable<Row>> sessionid2actionsRDD) {
		final String targetPageFlow = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW);
		final Long startPageId = Long.valueOf(targetPageFlow.split(",")[0]);
		final JavaRDD<Long> startPageRDD = sessionid2actionsRDD.flatMap(
				new FlatMapFunction<Tuple2<String, Iterable<Row>>, Long>() {
					@Override
					public Iterable<Long> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {
						List<Long> list = new ArrayList<Long>();
						final Iterator<Row> iterator = tuple._2.iterator();
						while (iterator.hasNext()) {
							final Row row = iterator.next();
							final long pageId = row.getLong(3);
							if (pageId == startPageId) {
								list.add(pageId);
							}
						}
						return list;
					}
				});
		return startPageRDD.count();
	}

	private static JavaPairRDD<String, Integer> generateAndMatchPageSplit(JavaSparkContext sc,
	                                                                      JavaPairRDD<String, Iterable<Row>> sessionid2actionsRDD,
	                                                                      JSONObject taskParam) {
		//获取到用户指定的页面流,比如:2,5,3,7
		final String targetPageFlow = ParamUtils.getParam(taskParam, Constants.PARAM_TARGET_PAGE_FLOW);
		//进行广播提高性能
		final Broadcast<String> targetPageFlowBroadcast = sc.broadcast(targetPageFlow);
		return sessionid2actionsRDD.flatMapToPair(
				new PairFlatMapFunction<Tuple2<String, Iterable<Row>>, String, Integer>() {
					@Override
					public Iterable<Tuple2<String, Integer>> call(Tuple2<String, Iterable<Row>> tuple) throws Exception {
						//定义返回的list
						List<Tuple2<String, Integer>> list = new ArrayList<Tuple2<String, Integer>>();
						//获取当前session的迭代器
						final Iterator<Row> iterator = tuple._2.iterator();
						//获取用户指定的页面流
						final String[] targetPages = targetPageFlowBroadcast.value().split(",");
						//默认迭代器里session都是无序的,需要根据访问时间来排序
						List<Row> rows = new ArrayList<Row>(10);
						while (iterator.hasNext()) {
							rows.add(iterator.next());
						}
						Collections.sort(rows, new Comparator<Row>() {
							@Override
							public int compare(Row o1, Row o2) {
								final String actionTime1 = o1.getString(4);
								final String actionTime2 = o2.getString(4);
								final Date date1 = DateUtils.parseTime(actionTime1);
								final Date date2 = DateUtils.parseTime(actionTime2);
								return (int) (date1.getTime() - date2.getTime());
							}
						});
						//页面切片的生成,以及页面流的匹配
						Long lastPageId = null;
						for (Row row : rows) {
							final long pageId = row.getLong(3);
							if (lastPageId == null) {
								lastPageId = pageId;
								continue;
							}
							//生成一个页面切面
							String pageSplit = lastPageId + "_" + pageId;
							//对这个切片判断一下,是否在用户指定的页面流中
							for (int i = 1; i < targetPages.length; i++) {
								final String targetPageSplit = targetPages[i - 1] + "_" + targetPages[i];
								if (targetPageSplit.equals(targetPageSplit)) {
									list.add(new Tuple2<String, Integer>(pageSplit, 1));
									break;
								}
							}
							lastPageId = pageId;
						}
						return list;
					}
				});
	}

	private static JavaPairRDD<String, Row> getSessionid2actionRDD(JavaRDD<Row> actionRDD) {

		return actionRDD.mapToPair(new PairFunction<Row, String, Row>() {
			@Override
			public Tuple2<String, Row> call(Row row) throws Exception {
				final String sessionid = row.getString(2);
				return new Tuple2<String, Row>(sessionid, row);
			}
		});
	}
}
